package cn.com.yto56.cainiao.exponent.elasticsearch.service;

import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;

public final class TaskProCenter {
    private static final int REQ_COUNT = 16;
    private static final int DEFAULT_TIME_OUT = 3600;
    private static final ExecutorService TH_SERVICE = Executors.newFixedThreadPool(REQ_COUNT);
    private static final ConcurrentHashMap<Long, Task> TASK_CACHE = new ConcurrentHashMap<>();
    private static final AtomicLong TASK_ID = new AtomicLong(0);

    static {
        Executors.newSingleThreadExecutor().execute(() -> {
            while (true) {
                try {
                    TASK_CACHE.forEach((k, v) -> {
                        v.check();
                    });
                    Thread.sleep(1);
                } catch (Exception e) {
                    // ignore
                }
            }
        });
    }

    public static void execute(int timeOutSecond, Runnable r) {
        TH_SERVICE.execute(new Task(timeOutSecond, r));
    }

    public static void execute(Runnable r) {
        execute(DEFAULT_TIME_OUT, r);
    }

    public static void repeatExecute(int repeatSize, Runnable r) {
        for (int i = 0; i < repeatSize; i++) {
            execute(r);
        }
    }

    public static <T> Future<T> submit(Callable<T> c) {
        return TH_SERVICE.submit(c);
    }

    static class Task implements Runnable {

        private Thread thread;
        private long taskId;
        private Runnable task;
        private long start;
        private long timeout;

        Task(int second, Runnable r) {
            taskId = TASK_ID.incrementAndGet();
            timeout = second * 1000;
            task = r;
        }

        @Override
        public void run() {
            try {
                start = System.currentTimeMillis();
                thread = Thread.currentThread();
                TASK_CACHE.put(taskId, this);
                task.run();
            } finally {
                TASK_CACHE.remove(taskId);
            }
        }

        public void check() {
            if ((System.currentTimeMillis() - start) > timeout) {
                thread.interrupt();
                TASK_CACHE.remove(taskId);
            }
        }
    }
}
